package com.feidee.fdhadoop.configUpdater;

import com.alibaba.fastjson.JSON;
import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.constant.CommonConstant;
import com.feidee.fdhadoop.constant.Constant;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.state.ConnectionState;
import org.apache.curator.framework.state.ConnectionStateListener;
import org.apache.curator.retry.RetryForever;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.data.Stat;
import org.slf4j.LoggerFactory;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;

public class ConfigUpdater {

    protected static final org.slf4j.Logger logger = LoggerFactory.getLogger(ConfigUpdater.class);

    private static ConfigUpdater cp = null;

    private String appPath;
    private CuratorFramework client;
    private ConfigUpdaterCallback configUpdaterCallback = null;

    private ConfigUpdater(final String appCode, ConfigUpdaterCallback configUpdaterCallback) {

        this.configUpdaterCallback = configUpdaterCallback;

        try {
            this.appPath = CommonConstant.ROOT_PATH + "/" + appCode;
            //初始化zk客户端，最多等待30秒
            CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
                try {
                    logger.info("ConfigUpdater initClient start~");
                    initClient(appCode);
                    logger.info("ConfigUpdater initClient end~");
                } catch (Exception e) {
                    logger.error("初始化zk客户端异常~", e);
                }
            });
            future.get(60, TimeUnit.SECONDS);
        } catch (Exception e) {
            logger.error("初始化ConfigProxy异常~", e);
        }
    }

    // 连接断开处理
    ConnectionStateListener connectionStateListener = new ConnectionStateListener() {

        public void stateChanged(CuratorFramework client, ConnectionState newState) {
            logger.info("listener connectionState:" + newState.name());
            if (newState.equals(ConnectionState.RECONNECTED)) {
                logger.info("重连后刷新数据和watch~");
                getConfigData(appPath, true);
            }
        }
    };

    private void initClient(String appCode) throws Exception {

        String env = CustomConfiguration.getString(CommonConstant.ENV_KEY, CommonConstant.ENV_LIVE);
        String zkConnectString = Constant.CONFIG_UPDATER_ZK_URL_LIVE;
        if (CommonConstant.ENV_TEST.equals(env)) {
            zkConnectString = Constant.CONFIG_UPDATER_ZK_URL_TEST;
        }

        RetryPolicy rf = new RetryForever(CommonConstant.RETRY_INTERVAL_MS);
        client = CuratorFrameworkFactory.newClient(zkConnectString, rf);
        client.start();

        client.getConnectionStateListenable().addListener(connectionStateListener);

        // 如果当前应用的znode尚未创建，则进行监听，待其创建后获取
        Stat s = client.checkExists().usingWatcher(appNodeWatcher).forPath(appPath);
        if (s == null) {
            logger.info("app znode is not exists~");
        } else {
            logger.info("app znode is exists~");
            getConfigData(appPath, false);
        }
    }

    // app节点增删改事件处理
    Watcher appNodeWatcher = new Watcher() {
        public void process(WatchedEvent event) {
            if (event.getType() == EventType.NodeCreated || event.getType() == EventType.NodeDataChanged) {
                logger.info("app node created or dataChanged~");
                getConfigData(event.getPath(), true);
            } else if (event.getType() == EventType.NodeDeleted) {
                logger.info("app node deleted~");
                try {
                    client.checkExists().usingWatcher(appNodeWatcher).forPath(appPath);// 设置exists监听，以接收appnode创建事件
                } catch (Exception e) {
                    logger.error("设置exists监听异常~", e);
                }
            }
        }
    };

    @SuppressWarnings("unchecked")
    private void getConfigData(String appNodePath, boolean executeCallBack) {
        try {
            String data = new String(client.getData().usingWatcher(appNodeWatcher).forPath(appNodePath));
            logger.info("data:" + data);
            CommonConstant.config = (ConcurrentHashMap<String, String>) JSON.parseObject(data, ConcurrentHashMap.class);
            logger.info("after getConfigData:" + JSON.toJSONString(CommonConstant.config));
            if (configUpdaterCallback != null && executeCallBack) {
                configUpdaterCallback.configUpdated(CommonConstant.config);
            }
        } catch (Exception e) {
            logger.error("获取config数据异常~", e);
        }
    }

    /**
     * 获取配置value
     *
     * @param key
     * @param defalutValue key不存在时返回的默认值
     */
    public String getConfig(String key, String defalutValue) {
        if (CommonConstant.config.containsKey(key)) {
            return CommonConstant.config.get(key);
        } else {
            return defalutValue;
        }
    }

    /**
     * 直接执行回调，初次获取配置后调用
     */
    public void executeCallBack() {
        this.configUpdaterCallback.configUpdated(CommonConstant.config);
    }

    /**
     * 单例，获取实例
     */
    public static ConfigUpdater getInstance(String appCode, ConfigUpdaterCallback configUpdaterCallback) {
        if (cp == null) {
            synchronized (ConfigUpdater.class) {
                if (cp == null) {
                    cp = new ConfigUpdater(appCode, configUpdaterCallback);
                }
            }
        }
        return cp;
    }
}
